package com.shenmazong.serverdemocanal.config;


import com.alibaba.otter.canal.client.CanalConnector;
import com.alibaba.otter.canal.client.CanalConnectors;
import com.alibaba.otter.canal.protocol.CanalEntry;
import com.alibaba.otter.canal.protocol.Message;
import com.google.protobuf.InvalidProtocolBufferException;
import jakarta.annotation.PostConstruct;
import org.springframework.stereotype.Component;

import java.net.InetSocketAddress;
import java.util.List;

/**
 * @author 军哥
 * @version 1.0
 * @description: TODO
 * @date 2025/11/6 14:36
 */

@Component
public class CanalClient {

    @PostConstruct
    public void run() {
        // 创建连接
        CanalConnector connector = CanalConnectors.newSingleConnector(
                new InetSocketAddress("192.168.80.192", 11111), "example", "canal", "canal");

        int batchSize = 1000;
        int emptyCount = 0;
        try {
            connector.connect();
            connector.subscribe("db_hotel\\.tb_user"); // 正则匹配
            connector.rollback(); // 回滚到未确认位置

            while (true) {
                Message message = connector.getWithoutAck(batchSize);
                long batchId = message.getId();
                int size = message.getEntries().size();

                if (batchId == -1 || size == 0) {
                    emptyCount++;
                    if (emptyCount > 10) {
                        Thread.sleep(1000);
                    }
                } else {
                    emptyCount = 0;
                    printEntry(message.getEntries());
                }

                connector.ack(batchId); // 确认消费
            }
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            connector.disconnect();
        }
    }

    private void printEntry(List<CanalEntry.Entry> entrys) {
        for (CanalEntry.Entry entry : entrys) {
            if (entry.getEntryType() == CanalEntry.EntryType.TRANSACTIONBEGIN
                    || entry.getEntryType() == CanalEntry.EntryType.TRANSACTIONEND) {
                continue;
            }

            CanalEntry.RowChange rowChg;
            try {
                rowChg = CanalEntry.RowChange.parseFrom(entry.getStoreValue());
            } catch (InvalidProtocolBufferException e) {
                throw new RuntimeException("解析 binlog 出错", e);
            }

            CanalEntry.EventType eventType = rowChg.getEventType();
            String tableName = entry.getHeader().getTableName();

            for (CanalEntry.RowData rowData : rowChg.getRowDatasList()) {
                if (eventType == CanalEntry.EventType.DELETE) {
                    // 处理删除
                    handleDelete(tableName, rowData);
                } else if (eventType == CanalEntry.EventType.INSERT) {
                    // 处理插入
                    handleInsert(tableName, rowData);
                } else if (eventType == CanalEntry.EventType.UPDATE) {
                    // 处理更新
                    handleUpdate(tableName, rowData);
                }
            }
        }
    }

    private void handleInsert(String tableName, CanalEntry.RowData rowData) {
        System.out.println("INSERT on table: " + tableName);
        rowData.getAfterColumnsList().forEach(col ->
                System.out.println(col.getName() + " = " + col.getValue()));
    }

    private void handleUpdate(String tableName, CanalEntry.RowData rowData) {
        System.out.println("UPDATE on table: " + tableName);
        System.out.println("Before: ");
        rowData.getBeforeColumnsList().forEach(col ->
                System.out.println(col.getName() + " = " + col.getValue()));
        System.out.println("After: ");
        rowData.getAfterColumnsList().forEach(col ->
                System.out.println(col.getName() + " = " + col.getValue()));
    }

    private void handleDelete(String tableName, CanalEntry.RowData rowData) {
        System.out.println("DELETE on table: " + tableName);
        rowData.getBeforeColumnsList().forEach(col ->
                System.out.println(col.getName() + " = " + col.getValue()));
    }
}